package com.bw.app.functions;

import com.alibaba.fastjson.JSONObject;
import com.bw.common.GmallConfig;
import com.bw.utils.DimUtil;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.Collection;
import java.util.Set;

public class DimFunction extends RichSinkFunction<JSONObject> {

    private Connection connection;

    @Override
    public void open(Configuration parameters) throws Exception {
        Class.forName(GmallConfig.PHOENIX_DRIVER);
        connection = DriverManager.getConnection(GmallConfig.PHOENIX_SERVER);
        connection.setAutoCommit(true);
    }

    @Override
    public void invoke(JSONObject value, Context context) throws Exception {
        JSONObject after = value.getJSONObject("after");//要插入的数据 {”id“:1,"tm_name":"XXX"}
        String sinkTable = value.getString("sinkTable");
        String upsertSql = getUpsertSql(after, sinkTable);
        System.out.println("upsertSql = " + upsertSql);

        String op = value.getString("type");
        // 删除redis  改了
        if ("update".equals(op)) {
            String id = after.getString("id");
            String key = "DIM:" + sinkTable.toUpperCase() + ":" + id;
            DimUtil.deleteDimForRedis(key);
        }
        PreparedStatement preparedStatement = null;
        try {
            // 预编译SQL
            preparedStatement = connection.prepareStatement(upsertSql);
            // 执行
            preparedStatement.executeUpdate();
        } catch (SQLException throwables) {
            throwables.printStackTrace();
        } finally {
            if (preparedStatement != null) {
                preparedStatement.close();
            }
        }


    }

    // 创建SQL
    // upsert into GMALL2105_REALTIME.dim_base_trademark(id,tm_name) values('1','华为')
    // after = {”id“:1,"tm_name":"XXX"}
    private String getUpsertSql(JSONObject after, String sinkTable) {
        Set<String> keys = after.keySet();//["id","name"]---id,tm_name
        Collection<Object> values = after.values();
        return "upsert into " + GmallConfig.HBASE_SCHEMA + "." + sinkTable + "("
                + StringUtils.join(keys, ",") + ")  values" + "('"
                + StringUtils.join(values, "','") + "')";
    }
}
